package com.learn.batch;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @create: 2023-04-17 15:53
 * @author: Mr.Du
 * --------------
 * @notes:
 **/
public class BatchWordCount {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //从文件中读取数据
        DataSource<String> input = env.readTextFile("./data/input/wordcount.txt");

        //对数据进行拆分
        input.flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String line, Collector<String> out) throws Exception {
                        String[] words = line.split(" ");
                        for (String word : words) {
                            //输出
                            out.collect(word);
                        }
                    }
                }).map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String word) throws Exception {

                        return Tuple2.of(word,1);
                    }
                })
                .groupBy(0)
                .sum(1).print();

    }

}
